package com.blyat.xsoft.kernel.component.mq.rocketmq.producer;

import com.blyat.xsoft.kernel.component.mq.MQListener;
import com.blyat.xsoft.kernel.component.mq.MQProducer;
import com.blyat.xsoft.kernel.component.mq.rocketmq.Carrier;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.util.Collection;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * @Auther: syh
 * @Date: 2020/8/11
 * @Description:
 */
@Component
@ConditionalOnProperty(prefix = "rocketmq", value = "name-server")
public class RocketMQProducer implements MQProducer {
    private static final Logger log = LoggerFactory.getLogger(RocketMQProducer.class);
    private static final int DEFAULT_TIME_OUT = 2 * 1000;
    private static final String DEFAULT_HASH_KEY = "XSOFT_HASH_KEY";

    @Autowired
    private RocketMQTemplate template;

    @Override
    public int send(String topic, Object content) {
        SendResult result = template.syncSend(topic, Carrier.build(content));
        return result.getSendStatus() != SendStatus.SEND_OK ? -1 : 0;
    }

    @Override
    public int send(String topic, Collection<Object> contents) {
        if (!Objects.isNull(contents) && !contents.isEmpty()) {
            Collection<Message> messages = contents.parallelStream()
                    .map(c -> MessageBuilder.withPayload(Carrier.build(c)).build())
                    .collect(Collectors.toList());

            SendResult result = template.syncSend(topic, messages);
            return result.getSendStatus() != SendStatus.SEND_OK ? -1 : 0;
        }
        return 0;
    }

    @Override
    public void sendAsync(String topic, Object content, MQListener asyncListener) {
        sendAsync(topic, content, DEFAULT_TIME_OUT, asyncListener);
    }

    @Override
    public void sendAsync(String topic, Object content, int timeout, MQListener asyncListener) {
        template.asyncSend(topic, Carrier.build(content), new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                if (asyncListener != null) {
                    asyncListener.onSuccess(sendResult.getSendStatus().ordinal());
                    log.info("send async to rocketmq ok:[topic:{},result:{}]",
                            topic, sendResult.getSendStatus().ordinal());
                }
            }

            @Override
            public void onException(Throwable e) {
                if (asyncListener != null) {
                    asyncListener.onException(e);
                }
                log.error("send async to rocketmq fail:[topic:{},error:{}]", topic, e);
                e.printStackTrace();
            }
        }, timeout);
    }

    @Override
    public void sendOneWay(String topic, Object content) {
        template.sendOneWay(topic, content);
    }

    @Override
    public int sendOrderly(String topic, Object content) {
        SendResult result = template.syncSendOrderly(topic, Carrier.build(content), DEFAULT_HASH_KEY);
        return result.getSendStatus() != SendStatus.SEND_OK ? -1 : 0;
    }

    @Override
    public int sendInTransaction(String topic, Object content) {
        Message message = MessageBuilder.withPayload(Carrier.build(content)).build();
        TransactionSendResult result = template.sendMessageInTransaction(topic, message, null);
        return result.getSendStatus() != SendStatus.SEND_OK ? -1 : 0;
    }
}
